package com.jacquesh.state;

import com.jacquesh.model.CarInfo;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class StateDemo0001 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment =
                StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setStateBackend(new FsStateBackend("file:/C:/data"));
        DataStreamSource<String> stream = environment
                .socketTextStream("192.168.88.180", 8888);
        stream.flatMap(new RichFlatMapFunction<String, CarInfo>() {

            @Override
            public void flatMap(String s, Collector<CarInfo> collector) throws Exception {
                try {
                    String[] sArr = s.split(",");
                    collector.collect(CarInfo.builder().carId(sArr[0]).speed(Long.parseLong(sArr[1])).build());
                } catch (Exception ex) {
                    System.out.println(ex);
                }
            }
        }).keyBy(CarInfo::getCarId).map(new RichMapFunction<CarInfo, CarInfo>() {

            private ReducingState<Long> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ReducingStateDescriptor lastTempStateDesc = new ReducingStateDescriptor("lastTempState", new ReduceFunction<Long>() {
                    @Override
                    public Long reduce(Long t1, Long t2) throws Exception {
                        return t1+t2;
                    }
                },
                BasicTypeInfo.LONG_TYPE_INFO);
                valueState = getRuntimeContext().getReducingState(lastTempStateDesc);
            }
            @Override
            public CarInfo map(CarInfo carInfo) throws Exception {
                valueState.add(carInfo.getSpeed());
                carInfo.setSpeed(valueState.get());
                return carInfo;
            }
        }).print(">>>>>>>>");

        environment.execute("execute");
    }
}
